package org.lql.chaos.kafka.example.design;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 14:34
 * @description:
 **/
@Component
public class KafkaTemplateManager {

	@Autowired
	@Qualifier("kafkaTemplateJson")
	private KafkaTemplate<String, Object> kafkaTemplateJson;

	@Autowired
	@Qualifier("kafkaTemplateKryo")
	private KafkaTemplate<String, Object> kafkaTemplateKryo;

	@Autowired
	@Qualifier("kafkaTemplateString")
	private KafkaTemplate<String, String> kafkaTemplateString;

	@Autowired
	@Qualifier("kafkaTemplateBytes")
	private KafkaTemplate<String, byte[]> kafkaTemplateBytes;

	public <K, V> KafkaTemplate<K, V> getKafkaTemplate(String serializationType) {
		switch (serializationType.toLowerCase()) {
			case "json":
				return (KafkaTemplate<K, V>) kafkaTemplateJson;
			case "kryo":
				return (KafkaTemplate<K, V>) kafkaTemplateKryo;
			case "string":
				return (KafkaTemplate<K, V>) kafkaTemplateString;
			case "bytes":
				return (KafkaTemplate<K, V>) kafkaTemplateBytes;
			default:
				throw new IllegalArgumentException("Unsupported serialization type: " + serializationType);
		}
	}


}
